package com.zisu.secs.secs.hsmsss;

import java.io.IOException;
import java.net.SocketAddress;
import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;

import com.zisu.secs.secs.SecsException;

/**
 * This abstract class is implementation of HSMS-SS-Active-Communicator(SEMI-E37.1).
 * 
 * <p>
 * This class is called from {@link HsmsSsCommunicator#newInstance(HsmsSsCommunicatorConfig)
 * </p>
 * 
 * @author kenta-shimizu
 *
 */
public abstract class AbstractHsmsSsActiveCommunicator extends AbstractHsmsSsCommunicator {

	public AbstractHsmsSsActiveCommunicator(HsmsSsCommunicatorConfig config) {
		super(Objects.requireNonNull(config));
	}
	
	@Override
	public void open() throws IOException {
		
		if ( hsmsSsConfig().protocol().get() != HsmsSsProtocol.ACTIVE ) {
			throw new IOException("HsmsSsCommunicatorConfig#protocol is not ACTIVE");
		}
		
		super.open();

		//循环指令 执行下面3个任务 ( 是通过这种方式实现断线 重连的，，当activeCircuit()其中一个环节出问题了，那么 立马clear 。然后 )
		executeLoopTask(() -> {
			
			activeCircuit();
			
			sendReplyManager.clear();
			
			hsmsSsConfig().timeout().t5().sleep();
		});
	}
	
	@Override
	public void close() throws IOException {
		super.close();
	}
	
	protected void activeCircuit() throws InterruptedException {
		
		try (
				AsynchronousSocketChannel channel = AsynchronousSocketChannel.open();
				) {
			
			final SocketAddress socketAddr = hsmsSsConfig().socketAddress().getSocketAddress();
			
			String socketAddrInfo = socketAddr.toString();
			
			try {
				//日志也是通过队列的形式 存储
				notifyLog("AbstractHsmsSsActiveCommunicator try-connect", socketAddrInfo);

				channel.connect(socketAddr, null, new CompletionHandler<Void, Void>(){

					//当channel 连接完成的时候
					@Override
					public void completed(Void none, Void attachment) {
						
						try {
							//也就是 只能由一个线程去连接 EQ 不能多个设备同时连
							if ( ! addChannel(channel) ) {
								return;
							}
							
							notifyHsmsSsCommunicateStateChange(HsmsSsCommunicateState.NOT_SELECTED);
							//阻塞队列 (BlockingQueue)是Java util.concurrent包下重要的数据结构，BlockingQueue提供了线程安全的队列访问方式：当阻塞队列进行插入数据时，如果队列已满，线程将会阻塞等待直到队列非满；从阻塞队列取数据时，如果队列已空，线程将会阻塞等待直到队列非空。
							//queue 放的数据是 设备主动发给HOST 的数据，并非是（data/linktest） 这种 host-》eq 后，host等待eq回复消息，，回复消息是不走queue，走的是回调
							final BlockingQueue<HsmsSsMessage> queue = new LinkedBlockingQueue<>();

							//reader也是一个线程任务
							final HsmsSsByteReader reader = new HsmsSsByteReader(AbstractHsmsSsActiveCommunicator.this, channel);
							final HsmsSsCircuitAssurance linktest = new HsmsSsCircuitAssurance(AbstractHsmsSsActiveCommunicator.this);

							//MessageReceiveListener 是唤醒 s1f18 的关键（也就是linktest-rsp 恢复的关键）
							//HsmsSsMessageReceiveListener() 这个方法返回的是 boolean ，但是参数是 个接口 需要我们在这里去去实现这个接口
							//todo here
							//实现的接口是 HsmsSsMessageReceiveListener里面的receive 接口，msg的参数类型是 HsmsSsMessage
							reader.addHsmsSsMessageReceiveListener(msg -> {
								//如果这个msg ，之前没有加入到 queue，那么把这个消息 ，加入到queue
								//System.err.println("activeCircuit-1"+msg+"hashCode:"+this.hashCode());
								//当所有msg进来以后，，会进入put 方法 ( offer: 将指定元素插入此队列中（如果立即可行且不会违反容量限制），成功时返回 true，如果当前没有可用的空间，则返回 false )
								//ifPresent(queue::offer) 如果不存在 说明之前队列里面没有，如果存在了put 方法里就会返回空
								sendReplyManager.put(msg).ifPresent(queue::offer);
							});

							//当有消息过来的时候，因为本次就是由消息通讯了，所以就不用和EQ进行Linktest 通讯了
							reader.addHsmsSsMessageReceiveListener(msg -> {
								System.err.println("activeCircuit-2"+msg+"hashCode:"+this.hashCode());
								linktest.reset();
							});
							
							final Callable<Void> mainTask = () -> {
								
								try {
									HsmsSsMessageSelectStatus ss = send(createSelectRequest())
											.map(HsmsSsMessageSelectStatus::get)
											//如果send() 返回的结果是null，那么将会执行orElse()里面的代码
											.orElse(HsmsSsMessageSelectStatus.NOT_SELECT_RSP);
									switch ( ss ) {
										case SUCCESS:
										case ACTIVED: {
										
											notifyHsmsSsCommunicateStateChange(HsmsSsCommunicateState.SELECTED);
											break;
										}
										default: {
											return null;
										}
									}
								}
								catch ( SecsException e ) {
									notifyLog(e);
									return null;
								}
								catch ( InterruptedException e ) {
									return null;
								}
								
								final Callable<Void> selectTask = () -> {
									try {
										//一直轮询 中queue中取出msg 来处理
										for ( ;; ) {
											HsmsSsMessage msg = queue.take();
											//获取Message Header 类型
											HsmsSsMessageType mt = HsmsSsMessageType.get(msg);
											try {
												switch ( mt ) {
													case DATA: {
														//如果收到了的是 数据类型 ，那么触发回调，这部分是业务 让业务程序去执行。
														notifyReceiveMessage(msg);
														break;
													}
													case SELECT_REQ: {
														//如果单单收到了SELECT_REQ 就说明有异常，一定是先发出SELECT_REQ再有SELECT_REQ
														send(createRejectRequest(msg, HsmsSsMessageRejectReason.NOT_SUPPORT_TYPE_S));
														break;
													}
													case LINKTEST_REQ: {
														send(createLinktestResponse(msg));
														break;
													}
													case SEPARATE_REQ: {
														return null;
														/* break; */
													}
													case SELECT_RSP:
													case DESELECT_REQ:
													case DESELECT_RSP:
													case LINKTEST_RSP:
													case REJECT_REQ:
													default: {
														/* ignore */
													}
												}
											}
											catch ( SecsException e ) {
												notifyLog(e);
											}
										}
									}
									catch ( InterruptedException ignore ) {
									}
									
									return null;
								};
								
								try {
									//invokeAny  当有一个任务完成之后（无论 成功/失败/异常 ），都会调用interrupt方法中断其它任务。
									executeInvokeAny(linktest, selectTask);
								}
								catch ( ExecutionException e ) {
									
									Throwable t = e.getCause();
									
									if ( t instanceof RuntimeException ) {
										throw (RuntimeException)t;
									}
									
									notifyLog(e);
								}
								catch ( InterruptedException ignore ) {
								}
								catch ( RejectedExecutionException e ) {
									if ( ! isClosed() ) {
										throw e;
									}
								}
								
								return null;
							};
							//一旦有1个任务正常完成(执行过程中没有抛异常)，线程池会终止其他未完成的任务
							//取得第一个完成任务的结果值，当第一个任务执行完成后，会调用interrupt方法将其他的任务中断。
							//因为正常情况下 reader 和 mainTask 是不会有返回值的是一直在后台跑，返回出来了 必定是出现异常了，所以要退出两个一起推出，然后开始下一次的连接启动
							executeInvokeAny(reader, mainTask);
						}
						catch ( ExecutionException e ) {
							
							Throwable t = e.getCause();
							
							if ( t instanceof RuntimeException ) {
								throw (RuntimeException)t;
							}
							
							if ( ! (t instanceof AsynchronousCloseException) ) {
								notifyLog(e);
							}
						}
						catch ( InterruptedException ignore ) {
						}
						catch ( RejectedExecutionException e ) {
							
							if ( ! isClosed() ) {
								throw e;
							}
						}
						finally {
							
							removeChannel(channel);
							
							try {
								channel.shutdownOutput();
							}
							catch (IOException giveup) {
							}
							
							synchronized ( channel ) {
								channel.notifyAll();
							}
						}
					}
					
					@Override
					public void failed(Throwable t, Void attachment) {
						
						synchronized ( channel ) {
							channel.notifyAll();
						}
						
						notifyLog("AbstractHsmsSsActiveCommunicator#open-AsynchronousSocketChannel#connect failed", t);
					}
				});
				
				synchronized ( channel ) {
					channel.wait();
				}
			}
			finally {
				notifyHsmsSsCommunicateStateChange(HsmsSsCommunicateState.NOT_CONNECTED);
			}
		}
		catch ( IOException e ) {
			notifyLog(e);
		}
	}

}
